/*
 * @Descripttion: 
 * @version: 
 * @Author: jiuguo.qin
 * @Date: 2020-01-02 10:34:27
 * @LastEditors: jiuguo.qin
 * @LastEditTime: 2020-05-17 19:18:12
 */

#include "wrk_core.h"
#include "script.h"
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>

#include <math.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <iostream>
#include <time.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/uio.h>
#include <netdb.h>
#include <inttypes.h>
#include "wrk_logger.h"
#include <sstream>
#include <sys/ioctl.h>
#include "include/ssl.h"
#include "server.h"
#include <exception>
#include "include/zmalloc.h"
#include "include/net.h"
#include "wrk_process.h"
#include "wrk_channel.h"
#include "wrk_worker.h"

#ifdef HAVE_AFFINITY
    #define _GNU_SOURCE
    #include <sched.h>
#endif


using namespace std;
struct http_parser_settings WrkCore::parser_request_settings = {
    on_message_begin: NULL,
    on_url  : on_request_url,
    on_status: NULL,
    on_header_field: on_request_header_field,
    on_header_value: on_request_header_value,
    on_headers_complete: NULL,
    on_body : on_request_body,
    on_message_complete : NULL,
    on_chunk_header : NULL,
    on_chunk_complete : NULL
};

struct http_parser_settings WrkCore::parser_response_settings = {
    on_message_begin: NULL,
    on_url  : NULL,
    on_status: NULL,
    on_header_field: NULL,
    on_header_value: NULL,
    on_headers_complete: NULL,
    on_body : NULL,
    on_message_complete : on_response_complete,
    on_chunk_header : NULL,
    on_chunk_complete : NULL
};

volatile sig_atomic_t WrkCore::stop = 1;

WrkRecord *WrkCore::record = NULL;
std::string WrkCore::script = std::string("");

wrk_core_config_t WrkCore::cfg = {
    connections : 0
};

void WrkCore::stop_handler(int sig) {
    WrkCore::stop = 1;
}

struct sock WrkCore::sock = {
    connect  : sock_connect, //非标准C，直接将回调函数赋值，对应epoll几个事件回调
    close    : sock_close,
    read     : sock_read,
    write    : sock_write,
    readable : sock_readable
};

WrkCore::WrkCore() {
    timecb_worker_thread_schedule_id = 0;
    // timecb_worker_thread_record_id = 0;
    stop = 1;
    thread_pool = NULL;
    cfg.timeout = 30 * 1000;
    record = new WrkRecord();
}

WrkCore::~WrkCore() {
    if(record) delete record;
}

wrk_core_thread_t* WrkCore::thread_pool = NULL;

// void WrkCore::set_stop_flag(sig_atomic_t flag)
// {
//     WrkCore::stop = flag;
// }

void WrkCore::set_sock(struct sock sk) {
    sock = sk;
}

char * WrkCore::copy_url_part(char *url, struct http_parser_url *parts, enum http_parser_url_fields field) {
    char *part = NULL;

    if (parts->field_set & (1 << field)) {
        uint16_t off = parts->field_data[field].off;
        uint16_t len = parts->field_data[field].len;
        part = (char *)malloc(len + 1 * sizeof(char));
		memset(part, 0 , len + 1 * sizeof(char));
        memcpy(part, &url[off], len);
    }

    return part;
}

int WrkCore::on_request_url(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *) parser->data;
    buffer_append(&c->request_path, at, len);
    *c->request_path.cursor++ = '\0';
    return 0;
}

int WrkCore:: on_request_header_field(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)parser->data;
    if (c->state == wrk_core_connection_t::VALUE) {
        *c->request_headers.cursor++ = '\0';
        c->state = wrk_core_connection_t::FIELD;
    }
    buffer_append(&c->request_headers, at, len);
    return 0;
}

int WrkCore:: on_request_header_value(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)parser->data;
    if (c->state == wrk_core_connection_t::FIELD) {
        *c->request_headers.cursor++ = '\0';
        c->state = wrk_core_connection_t::VALUE;
    }
    buffer_append(&c->request_headers, at, len);
    return 0;
}


int WrkCore:: on_request_body(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)parser->data;
    buffer_append(&c->request_body, at, len);
    *c->request_body.cursor++ = '\0';
    return 0;
}

int WrkCore::delay_request(aeEventLoop *loop, long long id, void *data) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)data;
    c->delayed = false;

    aeCreateFileEvent(loop, c->fd, AE_WRITABLE, socket_writeable, c);
    return AE_NOMORE;
}


void WrkCore::socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)data;
    
    if (stop) {
        aeStop(loop);
        return;
    }

    if (c->delayed) {
        return;
    }

    wrk_core_thread_t *t = c->t;
    if (!c->written) { //还未发出请求
        // 每发一次请求，都会运行LUA脚本中request函数构造出请求体
        script_request(t->L, c);
        if(c->request==NULL || c->request_length==0) {
            return;
        }

        //发送前解析出来
        script_verify_request(t->L, c);
        if(c->pipeline !=1) throw string("pipeline not support!");

        parse_request(c);
        if(c->request_path.size ==0 && cfg.is_force==false) throw string("request path parser failure!");
        c->start   = WrkRecord::time_us(); // 请求开始请求时间点，用来超时计时
    }

    char  *buf = c->request + c->written;
    size_t len = c->request_length  - c->written;
    size_t n;

    switch (sock.write(c, buf, len, &n)) {
        case OK:    break;
        case ERROR: goto error;
        case RETRY: return;
    }

    c->written += n;
    record->record_total.wbytes += n;

    if (c->written == c->request_length) {
        c->written = 0;
        record->record_total.requests++;
        // 关闭监听，在响应未处理过错成前，不再发送请求，响应处理response_complete开重新开启
        aeDeleteFileEvent(loop, fd, AE_WRITABLE);
    }

    record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].requests++;
    if(record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].ctime==0) {
        record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].ctime = WrkRecord::time_us();
    }
    return;

  error:
    record->record_total.errors.write++;
    record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].errors.write++;

    reconnect_socket(c);
}

void WrkCore::parse_request(wrk_core_connection_t *c) {
    //解析请求体，供控制流程用
    // buffer headers;
    http_parser *parser = (http_parser*)malloc(sizeof(http_parser)); // 分配一个http_parser
    parser->data = c; // 解析器回调把结果带出来，把结果写到connection对象上
    http_parser_init(parser, HTTP_REQUEST); // 初始化parser为Request类型
    c->state = wrk_core_connection_t::FIELD; //这个很重要

    http_parser_execute(parser, &parser_request_settings, c->request, c->request_length); //执行解析过程

    const char *method = http_method_str((enum http_method)parser->method);
    size_t len = strlen(method) + 1;
    char **buf = &c->request_method;
    *buf = (char *)realloc(*buf, len);
    memcpy(c->request_method, http_method_str((enum http_method)parser->method), len);

    free(parser);
    parser = NULL;
}

int WrkCore::on_response_complete(http_parser *parser) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)parser->data;
    wrk_core_thread_t *thread = c->t;

    int status = parser->status_code;

    string path = string(c->request_path.buffer, c->request_path.size);

    //请求数据及完成数累加
    record->record_total.complete++;
    if (status > 399) {
        record->record_total.errors.status++;
        record->map_tmp_data[path].errors.status++;
    }

    if (c->resp_headers.size>0) {
        *c->resp_headers.cursor++ = '\0'; //分隔出response header
        script_response(thread->L, status, c);
        c->state = wrk_core_connection_t::FIELD;
    }

    uint64_t elapsed_us = WrkRecord::time_us() - c->start;
    if (!record->stats_record_latency(elapsed_us)) {
        CDEBUG << "elapsed_us = " << elapsed_us <<endl;
        record->record_total.errors.timeout++;
        record->map_tmp_data[path].errors.timeout++;
    }else if(c->L) {
        lua_getglobal(c->L, "bizError");
        if(lua_type(c->L, -1)==LUA_TNUMBER) {
            // to do 
            record->map_tmp_data[path].errors.biz += lua_tointeger(c->L, -1);
            record->record_total.errors.biz++;
            lua_remove(c->L, -1);
        }
    }

    record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].complete++;
    record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].rt += elapsed_us;

    buffer_reset(&c->request_headers);
    buffer_reset(&c->request_path);
    buffer_reset(&c->request_body);
    buffer_reset(&c->resp_headers);
    buffer_reset(&c->resp_body);
    

    //重新开启监听写事件, socket_writeable时关掉了。
    if(!c->delayed) aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
    
    if (!http_should_keep_alive(parser)) {
        reconnect_socket(c);
        goto done;
    }

    // 复位，准备下个请求响应解析
    http_parser_init(parser, HTTP_RESPONSE);

  done:
    return 0;
}

int WrkCore::connect_socket(wrk_core_connection_t *c) 
{
    wrk_core_thread_t *thread = c->t;
    struct addrinfo *addr = thread->addr;
    struct aeEventLoop *loop = thread->loop;
    int fd, flags;

    fd = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);

    flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);

    if (connect(fd, addr->ai_addr, addr->ai_addrlen) == -1) {
        if (errno != EINPROGRESS) {
            LOG_ERROR("connect error: %s", strerror(errno));
            goto error;
        }
    }

    flags = 1;
    if(setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags)) ==-1){
        LOG_ERROR("setsockopt(fd=%d) error: %s", fd, strerror(errno));
        goto error;
    };

    //监听读写事件，可读/写时，触发回调socket_connected函数
    flags = AE_READABLE | AE_WRITABLE;
    if (aeCreateFileEvent(loop, fd, flags, socket_connected, c) == AE_OK) {
        c->parser.data = c; //将c附加到paser对象中，httpParser回调时从parser.data中解出connection
        c->fd = fd;
        return fd;
    }else {
        LOG_ERROR("eventLoop->setsize = %d, aeCreateFileEvent error: %s", loop->setsize, strerror(errno));
    }

  error:
    record->record_total.errors.connect++;
    close(fd);
    return -1;
}


void WrkCore::socket_connected(aeEventLoop *loop, int fd, void *data, int mask) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)data;

    switch (sock.connect(c, cfg.host)) {
        case OK:    break;
        case ERROR: goto error;
        case RETRY: return;
    }

    http_parser_init(&c->parser, HTTP_RESPONSE);
    c->written = 0;

    aeCreateFileEvent(c->t->loop, fd, AE_WRITABLE, socket_writeable, c);

    // 强攻模式忽略响应
    if(!cfg.is_force) aeCreateFileEvent(c->t->loop, fd, AE_READABLE, socket_readable, c);

    return;

  error:
    record->record_total.errors.connect++;
    reconnect_socket(c);
}

void WrkCore::socket_readable(aeEventLoop *loop, int fd, void *data, int mask) {
    wrk_core_connection_t *c = (wrk_core_connection_t *) data;
    size_t n;

    do {
        switch (sock.read(c, &n)) {
            case socket_status_t::OK:    break;
            case socket_status_t::ERROR: goto error;
            case socket_status_t::RETRY: return;
        }
        //计算网络读传输速率
        record->record_total.rbytes += n;

        // parser很复杂，用的高性能解析库。将c->buf按parser_settings解析
        // 结果放c->parser,如何从c->parser->data中解出connection对象？在connected中，将connection实例赋值到c->parser.data
        c->state = wrk_core_connection_t::FIELD;
        if ( http_parser_execute(&c->parser, &parser_response_settings, c->buf, n)!= n) {
            goto error; // 解析完成后进行回调
        }

        if (n == 0 && !http_body_is_final(&c->parser)) {
            goto error;
        } 
        
    } while (n == RECVBUF && sock.readable(c) > 0);

    return;

  error:
    record->record_total.errors.read++;
    record->map_tmp_data[string(c->request_path.buffer, c->request_path.size)].errors.read++;
    reconnect_socket(c);
}

int WrkCore::reconnect_socket(wrk_core_connection_t *c) {
    //销毁连接，删除文件事件，连接中会重新创建
    aeDeleteFileEvent(c->t->loop, c->fd, AE_WRITABLE | AE_READABLE);
    sock.close(c);
    close(c->fd);
    return connect_socket(c);
}

int WrkCore::on_response_header_field(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *) parser->data;
    if (c->state == wrk_core_connection_t::VALUE) {
        *c->resp_headers.cursor++ = '\0';
        c->state = wrk_core_connection_t::FIELD;
    }
    buffer_append(&c->resp_headers, at, len);
    return 0;
}

int WrkCore::on_response_header_value(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)parser->data;
    if (c->state == wrk_core_connection_t::FIELD) {
        *c->resp_headers.cursor++ = '\0';
        c->state = wrk_core_connection_t::VALUE;
    }
    buffer_append(&c->resp_headers, at, len);
    return 0;
}


int WrkCore::on_response_body(http_parser *parser, const char *at, size_t len) {
    wrk_core_connection_t *c = (wrk_core_connection_t *)parser->data;
    buffer_append(&c->resp_body, at, len);
    return 0;
}

int WrkCore::stop_test(aeEventLoop *loop, long long id, void *data)
{
    WrkCore::stop = 1;                              
    return AE_NOMORE;
}

size_t WrkCore::sock_can_readable(int fd) {
    int n, rc;
    rc = ioctl(fd, FIONREAD, &n);
    return rc == -1 ? 0 : n;
}


/**
 * @name: 
 * @test: 
 * @msg: 
 * @param {type} 
 * @return: 
 */
// int WrkCore::worker_thread_record(aeEventLoop *loop, long long id, void *data)
// {
//     record->add_point();
//     return WrkCore::stop==0 ? RECORD_INTERVAL_MS : AE_NOMORE;
// }


uint32_t WrkCore::threads_left;
pthread_mutex_t WrkCore::mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t WrkCore::cond = PTHREAD_COND_INITIALIZER;

/**
 * @name: 
 * @test: 
 * @msg: 
 * @param {type} 
 * @return: 
 */
int WrkCore::worker_thread_schedule(aeEventLoop *loop, long long id, void *data)
{
    pthread_mutex_lock(&mutex);
    model_ptr->phase_cursor++;
    threads_left = WrkCore::threads;
    if(model_ptr->phase_cursor==model_ptr->phase_cnt) {
        LOG_DEBUG("stop test, set stop=1");
        WrkCore::stop = 1;
        pthread_cond_broadcast(&cond);
        pthread_mutex_unlock(&mutex);
        return AE_NOMORE;
    }

    connections_left = model_ptr->phase[model_ptr->phase_cursor].connections;
    LOG_DEBUG("phase_cursor=%d, connections_left=%d", model_ptr->phase_cursor, connections_left);
    record->record_total.connections = connections_left;
    if(connections_left < WrkCore::threads) {
        wrk_core_thread_t *t = thread_pool;
        for(int i=0; i<WrkCore::threads - connections_left; i++, t++) {
            aeStop(t->loop);
        }
    }

    pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&mutex);
    return model_ptr->phase[model_ptr->phase_cursor].duration *1000;
}

void WrkCore::close_thread_connection(wrk_core_thread_t *t)
{
    wrk_core_connection_t *c = t->cs;
    for (uint64_t i = 0; i < t->connections; i++, c++) {
        close(c->fd);
        if(c->L) {
            lua_close(c->L);
            c->L = NULL;
        }
    }
}

uint64_t WrkCore::connections_left;

const uint32_t WrkCore::threads = WrkWorker::get_cpus_cores();

void *WrkCore::thread_main(void *arg) {
    pthread_detach(pthread_self());
    wrk_core_thread_t *t = (wrk_core_thread_t *)arg;
    #ifdef HAVE_AFFINITY
        cpu_set_t mask;
        CPU_ZERO(&mask);
        CPU_SET(t->seq, &mask); 
 
        if (pthread_setaffinity_np(pthread_self(), sizeof(mask),&mask) < 0) {
            LOG_WARNING("pthread_setaffinity_np failed: %s", strerror(errno));
        }else
        {
            LOG_INFO("pthread_setaffinity_np success");
        }
        
    #endif

    try
    {
        while(WrkCore::stop==0) 
        {
            pthread_mutex_lock(&mutex);
            
            while(connections_left<=0 && WrkCore::stop==0) {
                LOG_DEBUG("connections_left = %d, thread NO. %d pthread_cond_wait", connections_left, t->seq);
                pthread_cond_wait(&cond, &mutex);
            } 

            if(model_ptr->phase_cursor==model_ptr->phase_cnt) {
                pthread_mutex_unlock(&mutex);
                break;
            }
            
            uint64_t left_mod    = connections_left % threads_left;
            uint64_t threads_per = connections_left/threads_left; 
            t->connections = left_mod ==0 ? threads_per: threads_per + 1;
            connections_left -= t->connections;
            threads_left--;
            LOG_DEBUG("thread NO. %d, left_mod=%d, thread_per=%d, connections_left=%d, threads_left=%d", 
                        t->seq, left_mod, threads_per, connections_left, threads_left);
                        
            pthread_mutex_unlock(&mutex);
            
            wrk_core_connection_t *c = t->cs;
            for (uint64_t i = 0; i < t->connections; i++, c++) {
                c->L = luaL_newstate();
                if(c->L) luaL_openlibs(c->L);
                c->t = t;
                c->ssl     = cfg.ctx ? SSL_new(cfg.ctx) : NULL;
                c->request_method = NULL;
                c->start = 0;
                connect_socket(c);
            }

            LOG_DEBUG("thread No. %d create %d connection done", t->seq, t->connections);

            t->start = WrkRecord::time_us(); // 线程启动时间
            aeMain(t->loop);

            WrkCore::close_thread_connection(t);
        } // end while
            
    }
    catch (string &str)
    {
        std::cout << str <<std::endl;
    }
    catch(const std::out_of_range& e)
    {
        std::cerr << e.what() << std::endl;
    }
    catch(const std::bad_exception& e)
    {
        std::cerr << e.what() << std::endl;
    }
    catch(const std::bad_alloc& e)
    {
        std::cerr << e.what() << std::endl;
    }
    catch(...) {
        LOG_ERROR("thread_main have exception error! unkown!");
        std::cerr << "thread_main have exception error! unkown!"<<std::endl;
    }

    aeDeleteEventLoop(t->loop);
    cfg.stop_timestamp = WrkRecord::time_us();                                                          
    kill(server::master_pid, SIGUSR1);      
    
    WrkCore::close_thread_connection(t);

    stop_test(NULL, 0, NULL);

    zfree(t->cs);
    t->cs = NULL;
    LOG_DEBUG("thread NO. %d, task done! release t->cs(%sB), rss = %s", 
            t->seq, format_binary(t->connections * sizeof(wrk_core_connection_t)), format_binary(zmalloc_get_rss()));

    return NULL;
}


wrk_core_model_t* WrkCore::model_ptr = NULL;

void WrkCore::init_envirment()
{
    cfg.wrk_core_task_state = wrk_core_config_t::RUNNING;
    cfg.stop_timestamp = 0;
    cfg.last_correct_complete_timestamp = WrkRecord::time_us();
    cfg.last_correct_completes = 0;
    cfg.start = WrkRecord::time_us();

    if(cfg.L) {
        lua_close(cfg.L);
        cfg.L = NULL;
    }

    if(record !=NULL) {
        delete record;
        record = NULL;
    }

    if(thread_pool !=NULL) {
        delete[] thread_pool;
        thread_pool = NULL;
    }

    if(model_ptr !=NULL) {
        delete model_ptr;
        model_ptr = NULL;
    }

    if(timecb_worker_thread_schedule_id > 0) aeDeleteTimeEvent(WrkWorker::worker_loop, timecb_worker_thread_schedule_id);

    // if(timecb_worker_thread_record_id > 0) aeDeleteTimeEvent(WrkWorker::worker_loop, timecb_worker_thread_record_id);

    LOG_DEBUG("envirment init done, task state is setted running...");
}

void WrkCore::do_test(const char *script)
{   

    LOG_DEBUG("rss = %s ", format_binary(zmalloc_get_rss()));
    init_envirment();

    cfg.L = script_create((char *)script);
    model_ptr =  script_get_model(cfg.L);

    std::stringstream ss;
    cfg.max_connection = 0;
    ss << "Test Modle" << std::endl
        << "Url: "<<model_ptr->url <<std::endl
        << "timeout: " << model_ptr->timeout << std::endl
        << "phase cnt: " << model_ptr->phase_cnt << std::endl;
    for(int i=0; i<model_ptr->phase_cnt; i++) {
        wrk_core_phase_t *p = &model_ptr->phase[i];
        ss << "\t{id="<<p->id<<", connection="<<p->connections <<", duration="<<p->duration<<"}"<<std::endl;
        if(p->connections > cfg.max_connection) cfg.max_connection = p->connections;
    }

    WrkCore::script = std::string(script);

    LOG_DEBUG(ss.str().c_str());

    cfg.timeout     = model_ptr->timeout;
    record = new WrkRecord();
    
    char *schema    = script_get_wrk_field(cfg.L, "scheme");
    char *host      = script_get_wrk_field(cfg.L, "host");
    char *port      = script_get_wrk_field(cfg.L, "port");
    char *service = port ? port : schema;

    if (!strncmp("https", schema, 5)) {
        if ((cfg.ctx = ssl_init()) == NULL) {
            fprintf(stderr, "unable to initialize SSL\n");
            ERR_print_errors_fp(stderr);
            return ;
        }
        static struct sock sock;
        sock.connect  = ssl_connect;
        sock.close    = ssl_close;
        sock.read     = ssl_read;
        sock.write    = ssl_write;
        sock.readable = ssl_readable;
        WrkCore::set_sock(sock);
    }

    cfg.host = host;

    if (!script_resolve(cfg.L, host, service)) {
        char *msg = strerror(errno);
        fprintf(stderr, "unable to connect to %s:%s %s\n", cfg.host, service, msg);
        lua_close(cfg.L);
        cfg.L = NULL;
        return;
    }

    thread_pool = new wrk_core_thread_t[threads]();
    for(int i=0; i<threads; i++) {
        wrk_core_thread_t *t = &thread_pool[i];
        t->seq = i;
        t->L = script_create((char *)script);
        script_init(cfg.L, t);
        t->loop = aeCreateEventLoop(15 + cfg.max_connection * 2);

        // 构造连接上下文
        t->cs = (wrk_core_connection_t *)zcalloc(cfg.max_connection * sizeof(wrk_core_connection_t));
        wrk_memzero(t->cs, cfg.max_connection * sizeof(wrk_core_connection_t));

        if (i == 0) {
            // LUA脚本中有定义delay函数，返回一个时间，会对所有线程及连接生效
            if (script_want_response(t->L)) {
                parser_response_settings.on_header_field = on_response_header_field;
                parser_response_settings.on_header_value = on_response_header_value;
                parser_response_settings.on_body         = on_response_body;
            }
        }

        pthread_create(&t->thread, NULL, &thread_main, t);
    }

    signal(SIGPIPE, SIG_IGN);
    LOG_INFO("ingnore SIGPIPE signal");

    CDEBUG<<"rss = "<< format_binary(zmalloc_get_rss()) << std::endl;

    model_ptr->phase_cursor = -1;

    timecb_worker_thread_schedule_id = aeCreateTimeEvent(WrkWorker::worker_loop, 1000, worker_thread_schedule, NULL, NULL);

    // timecb_worker_thread_record_id = aeCreateTimeEvent(WrkWorker::worker_loop, RECORD_INTERVAL_MS, worker_thread_record, NULL, NULL);

    // register stop signal
    struct sigaction sa;
    sa.sa_handler = stop_handler;
    sa.sa_flags = 0;
    sigfillset(&sa.sa_mask);
    sigaction(SIGINT, &sa, NULL);
    LOG_INFO("register stop signal(SIGINT)");
}

string WrkCore::get_stats() 
{
    if(record==NULL) record = new WrkRecord();
    return record->get_stats();
}